package com.seckill.goods.listener;


import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.bean.copier.CopyOptions;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.seckill.goods.dispatcher.CanalSkuDataDispatcher;
import com.seckill.goods.pojo.Sku;
import com.seckill.search.feign.SkuInfoFeign;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.lang.reflect.Method;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.stream.Collectors;

@Component
@Slf4j
public class SkuIndexListener {

    @Autowired
    private CanalSkuDataDispatcher canalSkuDataDispatcher;

    @KafkaListener(topics = "seckill_goods_tb_sku", groupId = "goods_sku_index")
    public void skuIndexListen(ConsumerRecord<String, String> record){
        String key = record.key();
        String value = record.value();
        log.info("监听到的数据为: " + "key:" + key + ", value:" + value);
        //解析数据
        JSONObject jsonObject = JSONUtil.parseObj(value);//value中包含实际数据data,也包含操作等数据type(DELETE,UPDATE)
        JSONArray jsonArray = jsonObject.getJSONArray("data");
        //JSONArray jsonArray = JSONUtil.parseArray(jsonObject.get("data"));
        List<Sku> skus = jsonArray.stream().map(json -> {
            String jsonStr = JSONUtil.toJsonStr(json);
            Map map = JSON.parseObject(jsonStr, Map.class);
            Sku sku = BeanUtil.mapToBean(map, Sku.class, true, CopyOptions.create());
            return sku;
        }).collect(Collectors.toList());

        String type = jsonObject.getStr("type").toLowerCase();
        //同步到ES中
        try {
            Method method = canalSkuDataDispatcher.getClass().getMethod(type, List.class);
            method.invoke(canalSkuDataDispatcher, skus);
        } catch (Exception e) {
            e.printStackTrace();
            log.info("同步数据到ES出现问题,数据:{}", skus);
        }
    }
}
